package com.zsy.flink.kafka;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * @Description:* iot-temp topic输入内容类似， flink,pk,pk,spark, 先统计为DataStream<Tuple2<String, Integer>>
 * 然后将DataStream<Tuple2<String, Integer>>转换为DataStream<String> ， 最后将结果写入到kafka中，结果为Kafka and Flink says: (flink,1),(pk,2),(spark,1)格式
 * @ClassName: TestKafka
 * @Author: Zhou ShiYang
 * @Date: 2021/8/26 15:48
 */
public class TestKafka {
    private static final String KAFKA_BROKERS = "192.168.199.129:9092";

    /**
     * 第一步将kafka topic  iot-temp 作为source添加到DataStream
     * 第二步读取topic的内容进行单词统计
     * 第三步将统计结果进行转换
     * 第四步将结果存入到kafka另外一个topic中。
     *
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        // 使用ParameterTool.fromArgs从命令行创建ParameterTool(比如--input hdfs:///mydata --elements 42)
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setGlobalJobParameters(parameterTool);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", KAFKA_BROKERS);
        // kafka groupId
        properties.setProperty("group.id", "fink-kafka-connector");
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // topic：iot-temp
        DataStream<String> messageStream = env.addSource(
                new FlinkKafkaConsumer<String>("iot-temp", new SimpleStringSchema(), properties));

        DataStream<Tuple2<String, Integer>> counts = messageStream
                .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (String value, Collector<Tuple2<String, Integer>> out) -> {
                    // normalize and split the line
                    String[] splits = value.toLowerCase().split(",");
                    // emit the pairs
                    for (String s : splits) {
                        if (s.length() > 0) {
                            out.collect(new Tuple2<>(s, 1));
                        }
                    }
                })
                // 根据单词分组
                .keyBy(value -> value.f0)
                // 统计相同单词出现的个数
                .reduce((ReduceFunction<Tuple2<String, Integer>>) (Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) -> Tuple2.of(value1.f0, value1.f1 + value2.f1));
        counts.print();

        DataStream<String> countsString = counts.map((MapFunction<Tuple2<String, Integer>, String>) value -> {
            System.out.println("kafka msg = " + value);
            return "Kafka and Flink says: " + value;
        });

        // 发送到kafka
        FlinkKafkaProducer<String> myProducer = new FlinkKafkaProducer<>(KAFKA_BROKERS, "topic1", new SimpleStringSchema());
        ;

        myProducer.setWriteTimestampToKafka(true);
        countsString.addSink(myProducer);

        if (parameterTool.has("output")) {
            counts.writeAsText(parameterTool.get("output"));
        } else {
            System.out.println("Printing result to stdout. Use --output to specify output path.");
            counts.print();
        }

        // execute program
        JobExecutionResult result = env.execute("Streaming Kafka");
        JobID jobId = result.getJobID();
        System.out.println("jobId=" + jobId);
    }
}
